package com.distributed.kv.service;

import com.distributed.kv.dao.mapper.KvRepository;
import com.distributed.kv.dto.BatchRequest;
import com.distributed.kv.dto.BatchResponse;
import com.distributed.kv.dto.GroupByResponse;
import com.distributed.kv.dto.KvRequest;
import com.distributed.kv.dto.KvResponse;
import com.distributed.kv.entity.KvPair;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.time.LocalDate;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;
import java.util.Comparator;
import java.util.Map;

/**
 * KV核心业务服务
 *
 * 实现混合架构的核心逻辑:
 * 1. 查询流程: 缓存 -> 数据库 -> 分配新ID
 * 2. 写入流程: 数据库 -> 缓存更新
 * 3. 批量处理: 合并读写操作,提升性能
 * 4. 异步处理: 非阻塞操作,提高吞吐量
 */
@Service
@Transactional
public class KvService {

    private static final Logger logger = LoggerFactory.getLogger(KvService.class);

    private final KvRepository kvRepository;
    private final CacheService cacheService;
    private final IdGeneratorService idGeneratorService;

    @Value("${performance.batch.size:100}")
    private int batchSize;

    @Value("${performance.batch.timeout:5000}")
    private int batchTimeout;

    public KvService(KvRepository kvRepository,
                     CacheService cacheService,
                     IdGeneratorService idGeneratorService) {
        this.kvRepository = kvRepository;
        this.cacheService = cacheService;
        this.idGeneratorService = idGeneratorService;
    }

    /**
     * 获取或创建KV对
     * 核心查询流程实现
     */
    public KvResponse getOrCreate(String key) {
        return getOrCreate(key, null);
    }

    /**
     * 获取或创建KV对(带默认值)
     */
    public KvResponse getOrCreate(String key, String defaultValue) {
        if (key == null || key.trim().isEmpty()) {
            throw new IllegalArgumentException("Key cannot be null or empty");
        }

        try {
            // 1. 查询缓存
            KvPair kvPair = cacheService.get(key);
            if (kvPair != null) {
                return new KvResponse(kvPair.getKey(), kvPair.getId(), kvPair.getValue(), true);
            }

            // 2. 查询数据库
            Optional<KvPair> optional = kvRepository.findById(key);
            if (optional.isPresent()) {
                kvPair = optional.get();
                kvPair.incrementAccessCount();

                // 更新访问统计(异步)
                updateAccessCountAsync(key);

                // 回填缓存
                cacheService.put(key, kvPair);

                return new KvResponse(kvPair.getKey(), kvPair.getId(), kvPair.getValue(), true);
            }

            // 3. 创建新记录
            Long newId = idGeneratorService.nextId();
            kvPair = new KvPair(key, newId, defaultValue);

            // 保存到数据库
            kvRepository.save(kvPair);

            // 更新缓存
            cacheService.put(key, kvPair);

            logger.info("创建新KV对: {} -> {}", key, newId);
            return new KvResponse(key, newId, defaultValue, false);

        } catch (Exception e) {
            logger.error("获取或创建KV对失败: key={}", key, e);
            throw new RuntimeException("Failed to get or create KV pair: " + key, e);
        }
    }

    /**
     * 更新KV对的值
     */
    public KvResponse update(String key, String value) {
        if (key == null || key.trim().isEmpty()) {
            throw new IllegalArgumentException("Key cannot be null or empty");
        }

        try {
            Optional<KvPair> optional = kvRepository.findById(key);
            if (!optional.isPresent()) {
                throw new RuntimeException("Key not found: " + key);
            }

            KvPair kvPair = optional.get();
            kvPair.setValue(value);
            kvPair.setUpdatedAt(Instant.now());

            // 保存到数据库
            kvRepository.save(kvPair);

            // 更新缓存
            cacheService.put(key, kvPair);

            logger.info("更新KV对: {} -> {}", key, value);
            return new KvResponse(key, kvPair.getId(), value, true);

        } catch (Exception e) {
            logger.error("更新KV对失败: key={}, value={}", key, value, e);
            throw new RuntimeException("Failed to update KV pair: " + key, e);
        }
    }

    /**
     * 删除KV对
     */
    public boolean delete(String key) {
        if (key == null || key.trim().isEmpty()) {
            return false;
        }

        try {
            // 从数据库删除
            kvRepository.deleteById(key);

            // 从缓存删除
            cacheService.evict(key);

            logger.info("删除KV对: {}", key);
            return true;

        } catch (Exception e) {
            logger.error("删除KV对失败: key={}", key, e);
            return false;
        }
    }

    /**
     * 批量获取或创建
     */
    public BatchResponse<KvResponse> batchGetOrCreate(BatchRequest<String> request) {
        List<String> keys = request.getItems();
        if (keys == null || keys.isEmpty()) {
            return new BatchResponse<>(Collections.emptyList(), Collections.emptyList());
        }

        List<KvResponse> results = new ArrayList<>();
        List<String> errors = new ArrayList<>();

        // 分批处理,避免单次操作过大
        List<List<String>> batches = partitionList(keys, batchSize);

        for (List<String> batch : batches) {
            try {
                List<KvResponse> batchResults = processBatch(batch);
                results.addAll(batchResults);
            } catch (Exception e) {
                logger.error("批量处理失败: batch={}", batch, e);
                errors.add("Batch processing failed: " + e.getMessage());
            }
        }

        return new BatchResponse<>(results, errors);
    }

    /**
     * 异步批量处理
     */
    @Async
    public CompletableFuture<BatchResponse<KvResponse>> batchGetOrCreateAsync(BatchRequest<String> request) {
        return CompletableFuture.completedFuture(batchGetOrCreate(request));
    }

    /**
     * 获取热点数据统计
     */
    public List<KvResponse> getHotKeys(int limit) {
        try {
            List<KvPair> hotKeys = kvRepository.findHotKeys(10L); // 访问次数 > 10

            return hotKeys.stream()
                    .sorted((a, b) -> Long.compare(b.getAccessCount(), a.getAccessCount()))
                    .limit(limit)
                    .map(kv -> new KvResponse(kv.getKey(), kv.getId(), kv.getValue(), true))
                    .collect(Collectors.toList());

        } catch (Exception e) {
            logger.error("获取热点数据失败", e);
            return Collections.emptyList();
        }
    }

    /**
     * 缓存预热
     */
    public void warmUpCache(List<String> keys) {
        if (keys == null || keys.isEmpty()) {
            return;
        }

        try {
            List<KvPair> kvPairs = kvRepository.findByKeyIn(keys);

            for (KvPair kvPair : kvPairs) {
                cacheService.warmUp(kvPair.getKey(), kvPair);
            }

            logger.info("缓存预热完成,加载{}个Key", kvPairs.size());

        } catch (Exception e) {
            logger.error("缓存预热失败", e);
        }
    }

    // === 私有方法 ===

    private List<KvResponse> processBatch(List<String> keys) {
        // 1. 批量查询缓存
        Map<String, KvPair> cached = new HashMap<>();
        List<String> missingKeys = new ArrayList<>();

        for (String key : keys) {
            KvPair kvPair = cacheService.get(key);
            if (kvPair != null) {
                cached.put(key, kvPair);
            } else {
                missingKeys.add(key);
            }
        }

        // 2. 批量查询数据库
        Map<String, KvPair> fromDb = new HashMap<>();
        List<String> newKeys = new ArrayList<>();

        if (!missingKeys.isEmpty()) {
            List<KvPair> dbResults = kvRepository.findByKeyIn(missingKeys);
            Map<String, KvPair> dbMap = dbResults.stream()
                    .collect(Collectors.toMap(KvPair::getKey, kv -> kv));

            for (String key : missingKeys) {
                if (dbMap.containsKey(key)) {
                    fromDb.put(key, dbMap.get(key));
                } else {
                    newKeys.add(key);
                }
            }
        }

        // 3. 批量创建新记录
        Map<String, KvPair> newPairs = new HashMap<>();
        if (!newKeys.isEmpty()) {
            Long[] newIds = idGeneratorService.nextIds(newKeys.size());

            List<KvPair> toSave = new ArrayList<>();
            for (int i = 0; i < newKeys.size(); i++) {
                String key = newKeys.get(i);
                KvPair kvPair = new KvPair(key, newIds[i]);
                newPairs.put(key, kvPair);
                toSave.add(kvPair);
            }

            // 批量保存
            kvRepository.saveAll(toSave);
        }

        // 4. 更新缓存
        fromDb.values().forEach(kv -> cacheService.put(kv.getKey(), kv));
        newPairs.values().forEach(kv -> cacheService.put(kv.getKey(), kv));

        // 5. 构造结果
        List<KvResponse> results = new ArrayList<>();
        for (String key : keys) {
            KvPair kvPair = cached.get(key);
            boolean existed = true;

            if (kvPair == null) {
                kvPair = fromDb.get(key);
            }
            if (kvPair == null) {
                kvPair = newPairs.get(key);
                existed = false;
            }

            if (kvPair != null) {
                results.add(new KvResponse(key, kvPair.getId(), kvPair.getValue(), existed));
            }
        }

        return results;
    }

    @Async
    protected void updateAccessCountAsync(String key) {
        try {
            kvRepository.incrementAccessCount(key, Instant.now());
        } catch (Exception e) {
            logger.warn("更新访问计数失败: key={}", key, e);
        }
    }

    private <T> List<List<T>> partitionList(List<T> list, int size) {
        List<List<T>> partitions = new ArrayList<>();
        for (int i = 0; i < list.size(); i += size) {
            partitions.add(list.subList(i, Math.min(i + size, list.size())));
        }
        return partitions;
    }
    
    /**
     * 按值分组统计访问次数
     */
    public List<GroupByResponse.ValueStats> groupByValue(String valuePattern) {
        try {
            String pattern = valuePattern.contains("%") ? valuePattern : "%" + valuePattern + "%";
            List<Object[]> results = kvRepository.groupByValue(pattern);
            
            return results.stream()
                    .map(row -> new GroupByResponse.ValueStats(
                            (String) row[0],
                            (Long) row[1],
                            (Long) row[2]
                    ))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("按值分组查询失败: pattern={}", valuePattern, e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 按日期分组统计访问次数
     */
    public List<GroupByResponse.DateStats> groupByDate(LocalDate startDate, LocalDate endDate) {
        try {
            Instant startTime = startDate.atStartOfDay(ZoneId.systemDefault()).toInstant();
            Instant endTime = endDate.plusDays(1).atStartOfDay(ZoneId.systemDefault()).toInstant();
            
            List<KvPair> results = kvRepository.groupByDate(startTime, endTime);
            
            // 在Java中进行分组统计
            Map<LocalDate, List<KvPair>> groupedByDate = results.stream()
                    .collect(Collectors.groupingBy(kvPair -> 
                            kvPair.getCreatedAt().atZone(ZoneId.systemDefault()).toLocalDate()));
            
            return groupedByDate.entrySet().stream()
                    .map(entry -> {
                        LocalDate date = entry.getKey();
                        List<KvPair> pairs = entry.getValue();
                        long totalAccessCount = pairs.stream()
                                .mapToLong(pair -> pair.getAccessCount() != null ? pair.getAccessCount() : 0L)
                                .sum();
                        long recordCount = pairs.size();
                        return new GroupByResponse.DateStats(date, totalAccessCount, recordCount);
                    })
                    .sorted(Comparator.comparing(GroupByResponse.DateStats::getDate))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("按日期分组查询失败: startDate={}, endDate={}", startDate, endDate, e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 按日期和值分组统计访问次数
     */
    public List<GroupByResponse.DailyAccessStats> groupByDateAndValue(String valuePattern, LocalDate startDate, LocalDate endDate) {
        try {
            String pattern = valuePattern.contains("%") ? valuePattern : "%" + valuePattern + "%";
            Instant startTime = startDate.atStartOfDay(ZoneId.systemDefault()).toInstant();
            Instant endTime = endDate.plusDays(1).atStartOfDay(ZoneId.systemDefault()).toInstant();
            
            List<Object[]> results = kvRepository.groupByDateAndValue(pattern, startTime, endTime);
            
            return results.stream()
                    .map(row -> new GroupByResponse.DailyAccessStats(
                            ((java.sql.Date) row[0]).toLocalDate(),
                            (String) row[1],
                            (Long) row[2],
                            (Long) row[3]
                    ))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("按日期和值分组查询失败: pattern={}, startDate={}, endDate={}", valuePattern, startDate, endDate, e);
            return Collections.emptyList();
        }
    }
    
    /**
     * 查找包含特定值的记录
     */
    public List<KvResponse> findByValueContaining(String valuePattern) {
        try {
            String pattern = valuePattern.contains("%") ? valuePattern : "%" + valuePattern + "%";
            List<KvPair> results = kvRepository.findByValueContaining(pattern);
            
            return results.stream()
                    .map(kv -> new KvResponse(kv.getKey(), kv.getId(), kv.getValue(), true))
                    .collect(Collectors.toList());
        } catch (Exception e) {
            logger.error("按值查询失败: pattern={}", valuePattern, e);
            return Collections.emptyList();
        }
    }
}